/*
 * 文件名：RealDataDispatcher.java
 * 版权：亚信联创版权所有
 * 描述：TODO
 * 修改人：napo
 * 修改时间：2014-11-13
 * 修改内容：TODO
 */
package com.ailk.bigdata.etl.realstream.server.model;

import io.netty.channel.Channel;

import java.util.List;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.LinkedBlockingQueue;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.ailk.bigdata.etl.common.server.constant.Constant;
import com.ailk.bigdata.etl.realstream.server.tools.RealStreamTool;

/**
 * @description TODO
 * @author [xuwei3]
 * @version [版本号,2014-11-13]
 * @see [相关类/方法]
 * @since [产品/模块版本]
 */
public class RealDataDispatcher
{
	private static final Logger logger = LoggerFactory.getLogger(RealDataDispatcher.class);
	private LinkedBlockingQueue<RealStreamDataPack> SEND_DATA_QUEUE = new LinkedBlockingQueue<RealStreamDataPack>();
	private CopyOnWriteArrayList<NettyChannelInfo> SEND_CHANNEL_CACHE = new CopyOnWriteArrayList<NettyChannelInfo>();
	private boolean isPassiveSend = false;// 是否被动
	private volatile boolean isSendable = false;// 是否可发送
	private volatile boolean isRunning = false;// 是否正在运行
	private RealStreamUnit realStreamUnit;

	private RealDataDispatcher(RealStreamUnit streamUnit)
	{
		this.realStreamUnit = streamUnit;
	}

	public synchronized static RealDataDispatcher addDispatcherHand(RealStreamUnit streamUnit)
	{
		if (!RealStreamTool.DISPATCH_CAHCE.containsKey(streamUnit.getOnlyUnitCode()))
		{
			RealDataDispatcher dispatcher = new RealDataDispatcher(streamUnit);
			dispatcher.setPassiveSend(Constant.STREAM_TRANS_PASSIVE_MODE == streamUnit.getStreamTransMode());// 是否主动
			RealStreamTool.DISPATCH_CAHCE.put(streamUnit.getOnlyUnitCode(), dispatcher);
		}
		return RealStreamTool.DISPATCH_CAHCE.get(streamUnit.getOnlyUnitCode());
	}

	public static void addMessageQueue(RealStreamDataPack dataPack, RealStreamUnit streamUnit)
	{
		if (!RealStreamTool.DISPATCH_CAHCE.containsKey(streamUnit.getOnlyUnitCode()))
		{
			addDispatcherHand(streamUnit);
		}
		RealDataDispatcher dispatcher = RealStreamTool.DISPATCH_CAHCE.get(streamUnit.getOnlyUnitCode());
		dispatcher.addMessage(dataPack);
	}

	public void addMessage(RealStreamDataPack dataPack)
	{
		this.SEND_DATA_QUEUE.add(dataPack);
	}

	public RealStreamDataPack pollMessage()
	{
		return SEND_DATA_QUEUE.poll();
	}

	public RealStreamDataPack takeMessage() throws InterruptedException
	{
		return SEND_DATA_QUEUE.take();
	}

	public synchronized void addChannel(String channelId, Channel channel)
	{
		NettyChannelInfo channelInfo = new NettyChannelInfo();
		channelInfo.setChannel(channel);
		channelInfo.setRemoteIp(channelId);
		this.SEND_CHANNEL_CACHE.add(channelInfo);
		logger.info("数据端口[" + realStreamUnit.getOnlyUnitCode() + "]增加远程连接[" + channelId + "]成功");
	}

	public synchronized void rmChannel(String channelId, Channel channel)
	{
		for (int i = 0; i < SEND_CHANNEL_CACHE.size(); i++)
		{
			NettyChannelInfo channelInfo = this.SEND_CHANNEL_CACHE.get(i);
			if (channelInfo.getRemoteIp().equals(channelId))
			{
				this.SEND_CHANNEL_CACHE.remove(i);
				logger.info("数据端口[" + realStreamUnit.getOnlyUnitCode() + "]移除远程连接[" + channelId + "]成功");
			}
		}
	}

	public boolean isSendable()
	{
		return isSendable;
	}

	public void setSendable(boolean isSendable)
	{
		this.isSendable = isSendable;
	}

	public boolean isPassiveSend()
	{
		return isPassiveSend;
	}

	public void setPassiveSend(boolean isPassiveSend)
	{
		this.isPassiveSend = isPassiveSend;
	}

	public boolean isRunning()
	{
		return isRunning;
	}

	public void setRunning(boolean isRunning)
	{
		this.isRunning = isRunning;
	}

	public RealStreamUnit getRealStreamUnit()
	{
		return realStreamUnit;
	}

	public void setRealStreamUnit(RealStreamUnit realStreamUnit)
	{
		this.realStreamUnit = realStreamUnit;
	}

	public List<NettyChannelInfo> getSEND_CHANNEL_CACHE()
	{
		return SEND_CHANNEL_CACHE;
	}

}
